/**
 * log user info to table hbase
 * Main task log data 
 */

package olap4cloud.hbase;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorService;

import olap4cloud.data.DataStruct;

import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Increment;
import org.apache.hadoop.hbase.client.Put;

/**
 * @author  LuyenThanhDat
 */
public class DataToHbase implements Callable {

	private static final int MAXROWINBATCH = 500;
	private static final int MAX__NO__OF__BATCH = 500;

	/**
	 * @uml.property  name="objectCompletionService"
	 */
	private CompletionService<Object> objectCompletionService;
	/**
	 * @uml.property  name="eservice"
	 */
	private ExecutorService eservice;
	/**
	 * @uml.property  name="pool"
	 * @uml.associationEnd  multiplicity="(1 1)"
	 */
	private HTablePool pool;
	/**
	 * @uml.property  name="filePut"
	 */
	private String filePut;

	public static int TOTAL_LINE_INCRE = 0;

	public DataToHbase(HTablePool pool, String filePut) {
		this.pool = pool;
		this.filePut = filePut;
	}

	/**
	 * @return
	 * @uml.property  name="objectCompletionService"
	 */
	public CompletionService<Object> getObjectCompletionService() {
		return objectCompletionService;
	}

	/**
	 * @param objectCompletionService
	 * @uml.property  name="objectCompletionService"
	 */
	public void setObjectCompletionService(
			CompletionService<Object> objectCompletionService) {
		this.objectCompletionService = objectCompletionService;
	}

	/**
	 * @return
	 * @uml.property  name="eservice"
	 */
	public ExecutorService getEservice() {
		return eservice;
	}

	/**
	 * @param eservice
	 * @uml.property  name="eservice"
	 */
	public void setEservice(ExecutorService eservice) {
		this.eservice = eservice;
	}

	@Override
	public Object call() throws Exception {
		// TODO Auto-generated method stub

		int countNoOfBatch = 0;
		List<Put> listPut = new ArrayList<Put>();
		System.out.println(this.filePut);
		BufferedReader br = new BufferedReader(new FileReader(new File(this.filePut)));
		String line = null;
		int indext = 0;
		int count = 0;
		while ((line = br.readLine()) != null){
			indext ++;
			count++;
			DataStruct dataStruct = new DataStruct(line);
			
			Put put = TrackingTable.putDataToTable(indext, dataStruct);
			listPut.add(put);

			if (count >= MAXROWINBATCH) {
				
				TaskPutDataInToDB taskPut = new TaskPutDataInToDB(
						pool.getTable(TrackingTable.getTableName()),listPut );
				this.objectCompletionService.submit(taskPut);
				count = 0;
				listPut.clear();
				countNoOfBatch++;
				if (countNoOfBatch >= MAX__NO__OF__BATCH) {
					for (int i = 0; i < countNoOfBatch; i++) {
						Object o = objectCompletionService.take().get();
						countNoOfBatch = 0;
					}
				}
			}
		}

		if (count > 0) {
			TaskPutDataInToDB taskPut = new TaskPutDataInToDB(
					pool.getTable(TrackingTable.getTableName()),listPut );
			this.objectCompletionService.submit(taskPut);
			listPut.clear();
			count = 0;
		}

		for (int i = 0; i < countNoOfBatch; i++) {
			Object o = objectCompletionService.take().get();
		}

		System.out.println("Total line is : " + indext);
		return null;
	}

}
